package oc.tm.sg.core.monitor;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;

import oc.tm.sg.registry.common.utils.NamedThreadFactory;

import org.apache.log4j.Logger;

/**
 * 内存收集汇总,然后周期性发送
 * @author lry
 */
public class MemoryCollectMonitor implements Monitor {
    
    private static final Logger logger = Logger.getLogger(MemoryCollectMonitor.class);
    
    private static final int LENGTH = 6;
    
    // 定时任务执行器
    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3, new NamedThreadFactory("MemoryCollectMonitorSendTimer", true));

    // 统计信息收集定时器
    private final ScheduledFuture<?> sendFuture;
    private final Monitor monitor;
    private final long monitorInterval;
    
    // 获取过程锁
    private static final ReentrantLock LOCK = new ReentrantLock();
    private final ConcurrentMap<String, StatisticsSupport> statisticsMap = new ConcurrentHashMap<String, StatisticsSupport>();

    
    public MemoryCollectMonitor(Monitor monitor) {
        this.monitor = monitor;
        this.monitorInterval = 60000;//interval=60000
        // 启动统计信息收集定时器
        sendFuture = scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
            public void run() {
                // 收集统计信息
                try {
                    send();
                } catch (Throwable t) { // 防御性容错
                    logger.error("Unexpected error occur at send statistic, cause: " + t.getMessage(), t);
                }
            }
        }, monitorInterval, monitorInterval, TimeUnit.MILLISECONDS);
    }
    
    /**
     * 定时远程发送数据
     */
    public void send() {
        if (logger.isInfoEnabled()) {
            logger.info("Startting send statistics to monitor...");
        }
        long timestamp = System.currentTimeMillis();
        for (Map.Entry<String,StatisticsSupport> entry : statisticsMap.entrySet()) {
        	
        	StatisticsSupport st=entry.getValue();
            // 获取已统计数据
            Statistics statistics = st.getStatistics();
            AtomicReference<long[]> reference = st.getAtomicReference();
            long[] numbers = reference.get();
            long success = numbers[0];
            long failure = numbers[1];
            long elapsed = numbers[2];
            long concurrent = numbers[3];
            long maxElapsed = numbers[4];
            long maxConcurrent = numbers[5];
             
            // 发送汇总信息
            statistics.setInvoke_time(timestamp);	
            statistics.setSuccess(success);
            statistics.setFailure(failure);
            statistics.setElapsed(elapsed);
            statistics.setConcurrent(concurrent);
            statistics.setMax_elapsed(maxElapsed);
            statistics.setMax_concurrent(maxConcurrent);
            
            //远程推送数据
            monitor.collect(statistics);
            if (logger.isInfoEnabled()) {
                logger.info("Send statistics data to monitor:"+statistics.toString());
            }
            
            // 减掉已统计数据
            long[] current;
            long[] update = new long[LENGTH];
            do {
                current = reference.get();
                if (current == null) {//初始化为
                    update[0] = 0;
                    update[1] = 0;
                    update[2] = 0;
                    update[3] = 0;
                    update[4] = 0;
                    update[5] = 0;
                } else {//重置为0
                    update[0] = current[0] - success;
                    update[1] = current[1] - failure;
                    update[2] = current[2] - elapsed;
                    update[3] = current[3] - concurrent;
                    update[4] = current[4] - maxElapsed;
                    update[5] = current[5] - maxConcurrent;
                }
            } while (! reference.compareAndSet(current, update));
        }
    }
    
    /**
     * 内存收集汇总
     */
    public void collect(Statistics statistics) {
    	if (logger.isInfoEnabled()) {
            logger.info("Sg_Admin is recv statistics data from monitor: "+statistics.toString());
        }
    	
    	String key=statistics.getService()+"@"+statistics.getGroup()+"@"+statistics.getVersion();
    	
        // 读写统计变量
        long success = statistics.getSuccess();
        long failure = statistics.getFailure();
        long elapsed = statistics.getElapsed();
        long concurrent = statistics.getConcurrent();
        
        StatisticsSupport support=null;
        // 锁定
        LOCK.lock();
        try {
        	support=statisticsMap.get(key);
        } finally {
            // 释放锁
            LOCK.unlock();
        }

        // 初始化原子引用
        AtomicReference<long[]> reference = support.getAtomicReference();
        if (reference == null) {
            statisticsMap.putIfAbsent(key, new StatisticsSupport(statistics,new AtomicReference<long[]>()));
            reference = support.getAtomicReference();
        }
        // CompareAndSet并发加入统计数据
        long[] current;
        long[] update = new long[LENGTH];
        do {
            current = reference.get();
            if (current == null) {
                update[0] = success;
                update[1] = failure;
                update[2] = elapsed;
                update[3] = concurrent;
                update[4] = elapsed;
                update[5] = concurrent;
            } else {
                update[0] = current[0] + success;
                update[1] = current[1] + failure;
                update[2] = current[2] + elapsed;
                update[3] = (current[3] + concurrent) / 2;
                update[4] = current[4] > elapsed ? current[4] : elapsed;
                update[5] = current[5] > concurrent ? current[5] : concurrent;
            }
        } while (! reference.compareAndSet(current, update));
    }

    /**
     * 查找
     */
	public List<Statistics> lookup(Statistics query) {
		return monitor.lookup(query);
	}

	/**
	 * 销毁
	 */
    public void destroy() {
        try {
            sendFuture.cancel(true);
        } catch (Throwable t) {
            logger.error("Unexpected error occur at cancel sender timer, cause: " + t.getMessage(), t);
        }
    }

}